package com.blyat.xsoft.kernel.component.disruptor;

import com.lmax.disruptor.RingBuffer;

/**
 * @Auther: syh
 * @Date: 2020/7/9
 * @Description:
 */
public class MetadataProducer<T> implements Producer<T> {
    private String name;
    private RingBuffer<Metadata<T>> ringBuffer;
    private MetadataTranslator<T> translator;

    public MetadataProducer(String name) {
        this.name = name;
    }

    public MetadataProducer<T> setRingBuffer(RingBuffer<Metadata<T>> ringBuffer) {
        this.ringBuffer = ringBuffer;
        return this;
    }

    public MetadataProducer<T> setTranslator(MetadataTranslator<T> translator) {
        this.translator = translator;
        return this;
    }

    @Override
    public void produce(T t) {
        System.out.println(String.format("producer message by %s, data: %s", name, t));
        if (translator != null) {
            ringBuffer.publishEvent(translator, t);
        } else {
            long seq = ringBuffer.next();
            try {
                Metadata<T> metadata = ringBuffer.get(seq);
                metadata.set(t);
            } finally {
                // Disruptor 要求 RingBuffer.publish 必须得到调用的潜台词就是，如果发生异常也一样要调用 publish
                // 如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer
                ringBuffer.publish(seq);
            }
        }
    }
}
